package com.wsx.hbase;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.util.*;

public class HBaseExample {	
	static Configuration conf = HBaseConfiguration.create();
	private static String CONF_PATH= "/usr/local/hbase/hbase-1.2.5/conf/";
	static{
		conf.addResource(CONF_PATH+"hbase-site.xml");
	}
	//通过HBaseAdmin HTableDescriptor来创建一个新表
	public static void create(String tableName, String columnFamily) throws Exception{
		Connection con = ConnectionFactory.createConnection(conf);
		Admin admin = con.getAdmin();
		TableName tn = TableName.valueOf(tableName);
		if(admin.tableExists(tn)){
			delete(tn.toString());
		}
		HTableDescriptor table = new HTableDescriptor(tn);
		table.addFamily(new HColumnDescriptor(columnFamily).setCompactionCompressionType(Algorithm.NONE));
		admin.createTable(table);
		System.out.println("Table create success");
	}
	
	//添加一条数据，通过HTable Put为已存在的表添加数据
	public static void put(String tableName,String row,String columnFamily,String column,String data) throws IOException{
		Connection con = ConnectionFactory.createConnection(conf);
		TableName tn = TableName.valueOf(tableName);
		Table table = con.getTable(tn);
		Put put = new Put(Bytes.toBytes(row));
		put.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(column),Bytes.toBytes(data));
		table.put(put);
		System.out.println("put success");
	}
	
	//获取tableName表里列为row的结果集
	public static void get(String tableName,String row) throws IOException{
		Connection con = ConnectionFactory.createConnection(conf);
		Table table = con.getTable(TableName.valueOf(tableName));
		Get get = new Get(Bytes.toBytes(row));
		Result result = table.get(get);
		System.out.println("get "+ result);	
	}
	
	//通过HTable Scan来获取tableName表的所有数据信息
	public static void scan (String tableName) throws IOException{
		Connection con = ConnectionFactory.createConnection(conf);
		Table table = con.getTable(TableName.valueOf(tableName));
		Scan scan = new Scan();
		ResultScanner resultScanner = table.getScanner(scan);
		for(Result s : resultScanner){
			System.out.println("Scan "+ s);
		}
	}
	
	public static boolean delete(String tableName) throws Exception{
		Connection con = ConnectionFactory.createConnection(conf);
		Admin admin = con.getAdmin();
		TableName tn = TableName.valueOf(tableName);
		if(admin.tableExists(tn)){
			try {
				admin.disableTable(tn);
				admin.deleteTable(tn);
			} catch (Exception e) {
				e.printStackTrace();
				return false;
			}
		}
		return true;
	}
	
	public static void main(String[] args) {
		String tableName = "stu";
		String columnFamily = "iot";
		FileReader reader = null;
		BufferedReader br = null;
		try {
			HBaseExample.create(tableName, columnFamily);
			StringBuffer sb = new StringBuffer("");
			reader = new FileReader("doc/hbase/student");
			br = new BufferedReader(reader);
			String str = null;
			while ((str = br.readLine()) != null) {
				String[] strs = str.split(" ");
				HBaseExample.put(tableName, strs[1], columnFamily, "StudentNo", strs[0]);
				HBaseExample.put(tableName, strs[1], columnFamily, "name", strs[1]);
				HBaseExample.put(tableName, strs[1], columnFamily, "gender", strs[2]);
				HBaseExample.put(tableName, strs[1], columnFamily, "birthday", strs[3]);
				HBaseExample.put(tableName, strs[1], columnFamily, "class", strs[4]);
				HBaseExample.put(tableName, strs[1], columnFamily, "address", strs[5]);
				HBaseExample.put(tableName, strs[1], columnFamily, "scores", strs[6]);
				System.out.println(str);
			}
			
			
			HBaseExample.scan(tableName);	
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}